package me.flyray.bsin.server.listen;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import io.reactivex.disposables.Disposable;
import lombok.extern.slf4j.Slf4j;
import me.flyray.bsin.blockchain.listener.BsinBlockChainTransactionListenService;
import me.flyray.bsin.blockchain.transaction.BsinBlockChainTransactionProcessor;
import me.flyray.bsin.domain.entity.*;
import me.flyray.bsin.domain.enums.TransactionStatus;
import me.flyray.bsin.domain.request.TransactionDTO;
import me.flyray.bsin.enums.TransactionType;
import me.flyray.bsin.infrastructure.biz.TransactionBiz;
import me.flyray.bsin.infrastructure.biz.WalletAccountBiz;
import me.flyray.bsin.infrastructure.mapper.*;
import me.flyray.bsin.redis.provider.BsinCacheProvider;
import me.flyray.bsin.utils.BsinSnowflake;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import java.math.BigDecimal;
import java.math.BigInteger;
import java.time.LocalDateTime;
import java.util.*;

/**
 * 链上数据监听
 * 使用统一的区块链监听服务
 */
@Slf4j
@Service
public class ChainTransactionListen {

    @Resource
    ThreadPoolTaskExecutor taskExecutor;
    @Autowired
    private BsinBlockChainTransactionListenService transactionListenService;
    @Autowired
    private BsinBlockChainTransactionProcessor transactionProcessor;
    @Autowired
    private TransactionMapper transactionMapper;
    @Autowired
    private WalletAccountMapper walletAccountMapper;
    @Autowired
    private WalletMapper walletMapper;
    @Autowired
    private ChainCoinMapper chainCoinMapper;
    @Autowired
    private TransactionBiz transferBiz;
    @Autowired
    private WalletAccountBiz walletAccountBiz;
    
    // 存储活跃的监听订阅
    private final List<Disposable> activeSubscriptions = new ArrayList<>();


    /**
     * 启动链上交易监听
     */
    @PostConstruct
    public void executeAllMonitor() {
        try {
            log.info("启动链上交易监听服务");

            // 查询所有启用的合约
            QueryWrapper<ChainCoin> queryWrapper = new QueryWrapper<>();
            queryWrapper.eq("status", 1);
            queryWrapper.isNotNull("contract_address");
            queryWrapper.eq("coin", "EURO");
            
            List<ChainCoin> chainCoins = chainCoinMapper.selectList(queryWrapper);
            
            log.info("查询到 {} 个合约需要监听", chainCoins.size());
            for (ChainCoin coin : chainCoins) {
                log.debug("合约信息 - 币种: {}, 合约地址: {}, 链名称: {}", 
                    coin.getCoin(), coin.getContractAddress(), coin.getChainIdentifier());
            }

            if (chainCoins.isEmpty()) {
                log.info("没有找到需要监听的合约，跳过监听初始化");
                return;
            }

        // 按链分组合约
        Map<String, List<String>> chainContracts = new HashMap<>();
        for (ChainCoin chainCoin : chainCoins) {
            if (chainCoin.getChainIdentifier() != null) {  // 只处理有效的链名称
                chainContracts.computeIfAbsent(chainCoin.getChainIdentifier(), k -> new ArrayList<>())
                        .add(chainCoin.getContractAddress());
            }
        }

        // 为每条链启动监听
        for (Map.Entry<String, List<String>> entry : chainContracts.entrySet()) {
            String dbChainName = entry.getKey();
            List<String> contractAddresses = entry.getValue();
            
            // 映射数据库链名到配置链名
            String configChainName = mapDbChainNameToConfig(dbChainName);
            if (configChainName == null) {
                log.warn("跳过不支持的链: {}", dbChainName);
                continue;
            }

            log.info("为链 {} 启动 {} 个合约监听", dbChainName, contractAddresses.size());

            for (String contractAddress : contractAddresses) {
                try {
                    startContractListening(configChainName, contractAddress);
                } catch (Exception e) {
                    log.error("启动合约监听失败: chain={}, contract={}", configChainName, contractAddress, e);
                }
            }
        }

            log.info("链上交易监听服务启动完成，共启动 {} 个监听订阅", activeSubscriptions.size());
        } catch (Exception e) {
            log.error("链上交易监听服务启动失败", e);
        }
    }

    /**
     * 启动单个合约的监听
     */
    @Async("taskExecutor")
    public void startContractListening(String chainName, String contractAddress) {
        log.debug("开始监听智能合约：chain={}, contract={}", chainName, contractAddress);

        try {
            // 使用统一的交易监听服务
            Disposable subscription = transactionListenService.listenContractTransactions(
                    chainName, contractAddress, logEvent -> {
                        try {
                            // 使用统一的交易处理器处理交易事件
                            transactionProcessor.processTransactionEvent(chainName, logEvent, this::handleProcessedTransaction);
                        } catch (Exception e) {
                            log.error("处理交易事件失败: chain={}, contract={}, txHash={}",
                                    chainName, contractAddress, logEvent.getTransactionHash(), e);
                        }
                    }
            );

            if (!subscription.isDisposed()) {
                synchronized (activeSubscriptions) {
                    activeSubscriptions.add(subscription);
                }
                log.debug("成功启动合约监听: chain={}, contract={}", chainName, contractAddress);
            } else {
                log.warn("合约监听启动失败: chain={}, contract={}", chainName, contractAddress);
            }

        } catch (Exception e) {
            log.error("启动合约监听异常: chain={}, contract={}", chainName, contractAddress, e);
        }
    }

    /**
     * 处理已解析的交易 - 链上交易监听主入口
     * 这是用户充值流程的总入口点
     */
    private void handleProcessedTransaction(BsinBlockChainTransactionProcessor.ProcessedTransaction processedTx) {
        String txHash = processedTx.getTxHash();
        String chainName = processedTx.getChainName();

        log.info("处理交易: txHash={}, chain={}", txHash, chainName);

        try {
            log.debug("交易基础信息: txHash={}, contract={}, blockNumber={}, from={}, to={}", 
                    txHash, processedTx.getContractAddress(), processedTx.getBlockNumber(),
                    processedTx.getFromAddress(), processedTx.getToAddress());

            // 检查交易是否已经处理过 - 防重复处理
            QueryWrapper<Transaction> query = Wrappers.query();
            query.eq("tx_hash", txHash);
            List<Transaction> transactionList = transactionMapper.selectList(query);
            if (!transactionList.isEmpty()) {
                log.debug("交易已处理过，跳过: txHash={}", txHash);
                return;
            }

            // 使用缓存防并发重复处理
            String cachedValue = BsinCacheProvider.get("waas", txHash);
            if (cachedValue != null) {
                log.debug("交易正在处理中，跳过: txHash={}", txHash);
                return;
            }
            
            // 设置缓存锁
            BsinCacheProvider.put("waas", txHash, String.valueOf(processedTx.getBlockNumber()));

            try {
                // 获取合约对应的币种信息
                ChainCoin chainCoin = getChainCoinByContract(processedTx.getContractAddress());
                if (chainCoin == null) {
                    log.warn("未找到合约对应的币种信息，跳过处理: contract={}", processedTx.getContractAddress());
                    return;
                }
                log.debug("找到链币配置: contract={}, chainName={}, coinName={}", 
                        processedTx.getContractAddress(), chainCoin.getChainIdentifier(), chainCoin.getChainCoinName());

                // 创建交易记录
                TransactionDTO transactionDTO = createTransactionDTO(processedTx, chainCoin);

                // 处理转入交易 - 进入用户充值处理流程
                handleTransferIn(processedTx, transactionDTO, chainCoin);
                log.info("交易处理完成: txHash={}", txHash);

            } finally {
                // 释放缓存锁
                BsinCacheProvider.evict("waas", txHash);
            }

        } catch (Exception e) {
            log.error("处理交易失败: chain={}, txHash={}", chainName, txHash, e);
        }
    }

    /**
     * 创建交易DTO
     */
    private TransactionDTO createTransactionDTO(BsinBlockChainTransactionProcessor.ProcessedTransaction processedTx, ChainCoin chainCoin) {
        TransactionDTO transactionDTO = new TransactionDTO();
        transactionDTO.setTenantId(chainCoin.getTenantId());
        transactionDTO.setTxHash(processedTx.getTxHash());
        transactionDTO.setContractAddress(processedTx.getContractAddress());
        transactionDTO.setContractMethod(processedTx.getContractMethod());
        transactionDTO.setMethodInvokeWay(processedTx.getMethodInvokeWay());
        transactionDTO.setTransactionStatus("SUCCESS".equals(processedTx.getStatus()) ?
                TransactionStatus.SUCCESS.getCode() : TransactionStatus.FAIL.getCode());
        transactionDTO.setFromAddress(processedTx.getFromAddress());
        transactionDTO.setToAddress(processedTx.getToAddress());
        transactionDTO.setGasFee(new BigDecimal(processedTx.getGasUsed()));
        transactionDTO.setCompletedTime(LocalDateTime.now().toString());
        transactionDTO.setCreateTime(new Date());

        // 设置交易ID
        String serialNo = BsinSnowflake.getId();
        transactionDTO.setSerialNo(serialNo);

        // 转换金额：从 tokenAmount 获取
        BigInteger tokenAmount = processedTx.getTokenAmount();
        if (tokenAmount != null) {
            // 根据币种精度调整金额
            int decimals = chainCoin.getCoinDecimal() != null ? chainCoin.getCoinDecimal().intValue() : 18;
            
            // 检查精度值是否合理
            if (decimals < 0 || decimals > 255) {
                log.warn("币种精度值异常，使用默认值18: decimals={}", decimals);
                decimals = 18;
            }
            
            BigDecimal adjustedAmount = new BigDecimal(tokenAmount).divide(BigDecimal.TEN.pow(decimals));
            transactionDTO.setTxAmount(adjustedAmount);
        } else {
            log.warn("交易金额为空: txHash={}", processedTx.getTxHash());
        }
        return transactionDTO;
    }

    /**
     * 处理转入交易 - 用户充值流程
     * 流程步骤：
     * 1. 验证接收地址是否为平台生成的钱包
     * 2. 查找钱包账户和钱包配置信息
     * 3. 查询归集账户配置
     * 4. 保存交易记录
     * 5. 计算交易金额（按精度转换）
     * 6. 如果是充值钱包且需要归集：启动资金归集流程（包含gas提供）
     *    资金归集成功后会自动进行CRM余额充值和钱包余额更新
     * 7. 如果是非充值钱包，仅更新钱包账户余额
     */
    private void handleTransferIn(BsinBlockChainTransactionProcessor.ProcessedTransaction processedTx,
                                  TransactionDTO transactionDTO, ChainCoin chainCoin) throws Exception {

        String toAddress = processedTx.getToAddress();
        BigInteger tokenAmount = processedTx.getTokenAmount();

        log.info("处理用户充值转入交易: toAddress={}, amount={}, txHash={}", toAddress, tokenAmount, processedTx.getTxHash());

        // Step 1: 查找接收地址对应的钱包账户
        QueryWrapper<WalletAccount> toWalletAccountQueryWrapper = new QueryWrapper<>();
        toWalletAccountQueryWrapper.eq("address", toAddress);
        WalletAccount toWalletAccount = walletAccountMapper.selectOne(toWalletAccountQueryWrapper);

        if (toWalletAccount == null) {
            log.warn("接收地址不在系统中，跳过处理: toAddress={}", toAddress);
            return;
        }

        // Step 2: 查找钱包配置信息
        Wallet wallet = walletMapper.selectById(toWalletAccount.getWalletNo());
        if (wallet == null) {
            log.warn("未找到对应的钱包配置: walletNo={}", toWalletAccount.getWalletNo());
            return;
        }

        // Step 3: 查询归集账户配置
        ChainSettlementAccount gatherAccount = walletAccountBiz.getGatherAccount(wallet.getTenantId(), wallet.getTenantId() ,chainCoin.getSerialNo());

        if (gatherAccount == null) {
            log.warn("未找到归集账户配置，将跳过资金归集: tenantId={}, chainCoinSerialNo={}", 
                    wallet.getTenantId(), chainCoin.getSerialNo());
        }

        // Step 4: 确定交易类型
        if (gatherAccount != null && gatherAccount.getAccountNum().equals(toAddress)) {
            transactionDTO.setTransactionType(TransactionType.TRANSFER.getCode());
        } else {
            transactionDTO.setTransactionType(TransactionType.RECHARGE.getCode());
        }

        // Step 5: 设置交易信息并保存
        String serialNo = BsinSnowflake.getId();
        transactionDTO.setSerialNo(serialNo);
        transactionDTO.setBizRoleType(wallet.getBizRoleType());
        transactionDTO.setBizRoleTypeNo(wallet.getBizRoleTypeNo());
        transactionDTO.setTenantId(toWalletAccount.getTenantId());

        transactionMapper.insert(transactionDTO);

        // Step 6: 计算交易金额（按精度转换）
        int decimals = chainCoin.getCoinDecimal() != null ? chainCoin.getCoinDecimal().intValue() : 18;
        BigDecimal txAmount = new BigDecimal(tokenAmount).divide(BigDecimal.TEN.pow(decimals));

        // Step 7: 处理充值钱包的资金归集
        if (wallet.getWalletTag().equals("DEPOSIT") &&
                gatherAccount != null && !gatherAccount.getAccountNum().equals(toAddress)) {
            
            log.info("启动充值钱包资金归集流程: amount={}, toAddress={}", txAmount, toAddress);

            try {
                transferBiz.startUserFundCollectionProcess(
                        chainCoin.getChainIdentifier(),
                        toAddress,
                        gatherAccount.getAccountNum(),
                        processedTx.getContractAddress(),
                        tokenAmount,
                        chainCoin.getCoinDecimal()
                );
                log.info("资金归集流程启动成功: amount={}", txAmount);
            } catch (Exception e) {
                log.error("资金归集流程启动失败: amount={}, error={}", txAmount, e.getMessage(), e);
            }
            
        } else {
            // 非充值钱包类型，直接更新钱包账户余额
            BigDecimal oldBalance = toWalletAccount.getBalance();
            BigDecimal newBalance = oldBalance.add(txAmount);
            toWalletAccount.setBalance(newBalance);
            walletAccountMapper.updateById(toWalletAccount);
            log.info("非充值钱包余额更新: address={}, amount={}", toAddress, txAmount);
        }
        
        log.info("用户充值转入交易处理完成: txHash={}", processedTx.getTxHash());
    }

    /**
     * 映射数据库链名到配置链名
     */
    private String mapDbChainNameToConfig(String dbChainName) {
        switch (dbChainName.toUpperCase()) {
            case "ETHEREUM":
                return "ethereum";
            case "BSC":
                return "bsc";
            case "POLYGON":
                return "polygon";
            case "TRON":
                return "tron";
            case "CONFLUX":
                return "conflux";
            default:
                log.warn("未知的链名称: {}", dbChainName);
                return null;
        }
    }


    /**
     * 根据合约地址获取币种信息
     */
    private ChainCoin getChainCoinByContract(String contractAddress) {
        // 使用LambdaQueryWrapper确保类型安全
        LambdaQueryWrapper<ChainCoin> queryWrapper = new LambdaQueryWrapper<>();
        queryWrapper.eq(ChainCoin::getContractAddress, contractAddress);
        queryWrapper.eq(ChainCoin::getStatus, 1);
        
        ChainCoin result = chainCoinMapper.selectOne(queryWrapper);
        
        if (result == null) {
            log.warn("未找到链币配置: contract={}", contractAddress);
        }
        
        return result;
    }

    /**
     * 启动单个合约的监听（向后兼容方法）
     * @param contractAddress 合约地址
     */
    @Async("taskExecutor")
    public void contractAddressMonitor(String contractAddress) {
        // 根据合约地址推断链名称（这里简化处理，实际应该查询数据库）
        String chainName = "bsc"; // 默认使用 BSC
        startContractListening(chainName, contractAddress);
    }

    /**
     * 转出交易（向后兼容方法）
     */
    public void transferOut() {
        log.info("执行转出交易");
        // TODO: 实现转出交易逻辑
        // 这里可以调用具体的转出服务
    }

    /**
     * 停止所有监听服务
     */
    @PreDestroy
    public void stopListeners() {
        log.info("开始停止链上交易监听服务...");

        synchronized (activeSubscriptions) {
            for (Disposable subscription : activeSubscriptions) {
                if (!subscription.isDisposed()) {
                    subscription.dispose();
                }
            }
            activeSubscriptions.clear();
        }

        log.info("链上交易监听服务已停止");
    }
}
